package com.deepglint.source;

import com.deepglint.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

// 自定义的source
public class SourceTest_UDF {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 输出有序
        env.setParallelism(1);

        DataStream<SensorReading> dataSet = env.addSource(new MySensorSource());

        dataSet.print();

        env.execute();
    }

    // 自定义source，需要实现SourceFunction接口
    // 泛型指定返回的数据类型
    public static class MySensorSource implements SourceFunction<SensorReading> {

        // 定义标志位，来控制数据源的产生和关闭
        private boolean flag = true;

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            // 随机数生成器
            Random random = new Random();

            // 设置10个传感器的初始温度
            HashMap<String, Double> map = new HashMap<>();
            for (int i = 0; i < 10; i++) {
                map.put("sensor_" + (i), 60 + random.nextGaussian() * 20);
            }

            while (flag) {
                for (String sensorId : map.keySet()) {
                    // 在当前温度的基础上随机波动
                    Double newTemp = map.get(sensorId) + random.nextGaussian();
                    map.put(sensorId, newTemp);
                    ctx.collect(new SensorReading(sensorId, sensorId, System.currentTimeMillis(), newTemp));
                }

                // 控制输出频率
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}